import os
import asyncio
import threading
import time
import uuid
import queue
import logging
import pyaudio
import wave
from typing import Optional
from concurrent.futures import CancelledError as FuturesCancelledError
import redis
from api_client import set_guaduan_phone, send_sms, store_chat_data, is_sending_sms
from settings import REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_PASSWORD, \
    REDIS_TALK_STATUS_PREFIX, VOICE_CONNECTED_VALUE, INACTIVITY_TIMEOUT_SECONDS, DEVICE_PHONENUM
from doubao.realtime_dialog_client import RealtimeDialogClient
from doubao import config as doubao_config
from logging_config import get_device_logger

def set_device_context(device_phone: Optional[str]):
    """设置当前线程的设备上下文（包装函数）
    
    Args:
        device_phone: 设备手机号
    """
    from gui_multi_device import _set_device_context
    _set_device_context(device_phone)

# 命中这些关键词则直接置为 novice 并跳过业务调用
BLOCK_PHRASES = [
    "语音留言",
    "联系的用户无法接听"
]

# 音频参数（与豆包配置保持一致）
FORMAT = pyaudio.paInt16  # 16位深度
CHANNELS = 1              # 单声道
RATE = 16000              # 采样率 16kHz
CHUNK = 3200              # 每块大小（与豆包demo保持一致，3200帧 = 200ms @ 16kHz）



class RealTimeRecognizer:
    """基于豆包端对端实时语音的识别器"""
    
    def __init__(self, device_index: Optional[int] = None, device_phone: Optional[str] = None, 
                 output_device_index: Optional[int] = None):
        self.device_index = device_index
        self.device_phone = device_phone or DEVICE_PHONENUM
        self.output_device_index = output_device_index
        self.redis_key = f"{REDIS_TALK_STATUS_PREFIX}{self.device_phone}"
        # 设置设备上下文
        set_device_context(self.device_phone)
        self.logger = get_device_logger(self.device_phone)
        
        # 异步相关
        self.loop = None
        self.loop_thread = None
        self.client = None
        self.session_id = None
        
        # 状态管理
        self.running = False
        self.last_text_time = time.time()
        self.current_text = ""
        self.redis_client = None
        
        # 业务相关
        self.called_phone = None
        self.session_id_for_round = None
        
        # 音频队列（用于输出）
        self.audio_queue = queue.Queue()
        self.output_stream = None
        self.input_stream = None  # 确保stop()在未开启麦克风时也能安全执行
        self.audio = None
        self.output_format = 'float32'  # 默认Float32
        self.is_playing = False
        self.player_thread = None
        
        # 文本识别结果回调
        self.on_text_received = None
        self.is_user_querying = False
        self.say_hello_over_event = None  # 将在事件循环中初始化
        self.customer_info = None  # 存储客户信息，用于配置提示词
        
        # 音频保存相关
        self.audio_save_dir = None  # 音频保存目录
        self.input_audio_file = None  # 输入音频文件句柄
        self.output_audio_file = None  # 输出音频文件句柄
        self.input_audio_writer = None  # 输入音频wave writer
        self.output_audio_writer = None  # 输出音频wave writer
        
        # 豆包回复文本收集
        self.current_reply_text = ""  # 当前正在收集的回复文本（完整的）
        self.all_replies = []  # 所有完整的回复文本列表 [{session_id, user_text, doubao_text, timestamp}]

        #短信状态判断相关
        self.sms_doubao_flag = False
        
        # 开场白播放状态
        self.hello_playing = False            # 当前是否在播放开场白
        self.hello_streaming_done = True     # 服务端是否已推送完开场白音频
    
    def set_device_context(self, device_phone: Optional[str]):
        """设置设备上下文（用于日志路由）
        
        Args:
            device_phone: 设备手机号
        """
        set_device_context(device_phone)

    def _init_redis(self):
        """延迟初始化Redis连接"""
        if self.redis_client is None and redis is not None:
            try:
                self.redis_client = redis.Redis(
                    host=REDIS_HOST,
                    port=REDIS_PORT,
                    db=REDIS_DB,
                    password=REDIS_PASSWORD,
                    decode_responses=True,
                    socket_timeout=2,
                )
            except Exception as e:
                self._log("Redis 初始化失败: {}".format(e))
    
    def _init_audio(self):
        """延迟初始化音频对象"""
        if self.audio is None:
            try:
                self.audio = pyaudio.PyAudio()
            except Exception as e:
                self._log(f"音频初始化失败: {e}")
                self.audio = None

    def _log(self, message: str, level: int = logging.INFO):
        """设备感知的日志打印

        确保日志消息带有正确的设备上下文
        """
        # 确保当前线程的设备上下文正确
        set_device_context(self.device_phone)
        if self.logger:
            self.logger.log(level, message)
        else:
            logging.getLogger(__name__).log(level, message)

    def _finish_hello_playback(self):
        """当开场白音频真正播放结束时调用"""
        # 确保设备上下文正确
        set_device_context(self.device_phone)
        if self.hello_playing or  self.hello_streaming_done:
            self.hello_playing = False
            self.hello_streaming_done = True
            self.last_text_time = time.time()
            self._log("开场白播放完成,已恢复打断模式")
    
    def _setup_event_loop(self):
        """在独立线程中设置事件循环"""
        # 如果循环存在且正在运行，需要先停止它
        if self.loop is not None:
            if self.loop.is_running():
                try:
                    # 停止现有循环
                    self.loop.call_soon_threadsafe(self.loop.stop)
                    if self.loop_thread:
                        self.loop_thread.join(timeout=5)
                except:
                    pass
            # 重置循环相关变量
            self.loop = None
            self.loop_thread = None
            self.say_hello_over_event = None
        
        def run_loop():
            # 在事件循环线程中设置设备上下文
            set_device_context(self.device_phone)
            self.loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self.loop)
            # 初始化 say_hello_over_event
            self.say_hello_over_event = asyncio.Event()
            self.loop.run_forever()
        
        self.loop_thread = threading.Thread(target=run_loop, daemon=True)
        self.loop_thread.start()
        
        # 等待循环启动和事件初始化
        timeout = 5.0
        start_time = time.time()
        while (self.loop is None or self.say_hello_over_event is None) and (time.time() - start_time < timeout):
            time.sleep(0.1)
        
        if self.loop is None or self.say_hello_over_event is None:
            raise Exception("事件循环初始化失败或超时")
    
    def _run_in_loop(self, coro, timeout=300):
        """在事件循环中运行协程
        
        Args:
            coro: 要运行的协程
            timeout: 超时时间（秒），默认300秒（5分钟），因为对话可能较长
        """
        future = asyncio.run_coroutine_threadsafe(coro, self.loop)
        try:
            return future.result(timeout=timeout)
        except FuturesCancelledError:
            # 任务被正常取消（例如通话结束时主动取消子任务）
            return None
        except TimeoutError:
            self._log(f"运行协程超时（{timeout}秒），取消任务")
            future.cancel()
            # 尝试等待取消完成
            try:
                future.result(timeout=1)
            except:
                pass
            # raise
    
    async def _handle_server_response(self, response):
        """处理服务器响应"""
        # 确保设备上下文正确
        set_device_context(self.device_phone)

        if response == {}:
            return

        message_type = response.get('message_type')
        # print(response)
        # 处理音频数据（TTS输出） - 使用队列和播放线程，事件450时清空队列
        if message_type == 'SERVER_ACK' and isinstance(response.get('payload_msg'), bytes):
            audio_data = response['payload_msg']
            output_audio_bytes = None  # 用于保存的音频数据

            # Barge-in: 用户说话期间不播放也不保存（但如果正在播放开场白，则继续播放）
            if self.is_user_querying and not self.hello_playing:
                return

            try:
                # 确保播放器已启动（在某些设备顺序下，可能先收到音频后才建好线程）
                if self.output_stream and not self.is_playing:
                    self.is_playing = True
                    # 保存设备上下文给播放线程使用
                    device_phone = self.device_phone
                    def _player():
                        # 在播放线程中设置设备上下文
                        set_device_context(device_phone)
                        idle_loops = 0
                        while self.is_playing:
                            try:
                                data = self.audio_queue.get(timeout=1.0)
                                idle_loops = 0
                                if not data:
                                    continue
                                # 如果用户正在说话，且不在播放开场白，则跳过播放
                                if self.is_user_querying and not self.hello_playing:
                                    continue
                                if self.output_stream and not self.output_stream.is_active():
                                    try:
                                        self.output_stream.start_stream()
                                    except Exception:
                                        pass
                                try:
                                    self.output_stream.write(data)
                                except Exception:
                                    time.sleep(0.01)
                            except queue.Empty:
                                idle_loops += 1
                                # 如果开场白流式传输已结束，且队列已空闲，立即恢复（不需要等待2次）
                                if self.hello_playing and self.hello_streaming_done and idle_loops >= 1:
                                    self._finish_hello_playback()
                                time.sleep(0.05)
                            except Exception:
                                time.sleep(0.05)
                    self.player_thread = threading.Thread(target=_player, daemon=True)
                    self.player_thread.start()

                if self.output_format == 'int16':
                    try:
                        import numpy as np
                        audio_array = np.frombuffer(audio_data, dtype=np.float32)
                        audio_array = np.clip(audio_array, -1.0, 1.0)
                        audio_int16 = (audio_array * 32767).astype(np.int16)
                        output_audio_bytes = audio_int16.tobytes()
                        # 播放：放入队列
                        if self.is_playing:
                            self.audio_queue.put(output_audio_bytes)
                    except ImportError:
                        self._log("需要numpy库来转换音频格式，请安装: pip install numpy")
                    except Exception as e:
                        self._log(f"音频格式转换失败: {e}")
                else:
                    # Float32播放，另行生成Int16用于保存
                    try:
                        import numpy as np
                        audio_array = np.frombuffer(audio_data, dtype=np.float32)
                        audio_array = np.clip(audio_array, -1.0, 1.0)
                        audio_int16 = (audio_array * 32767).astype(np.int16)
                        output_audio_bytes = audio_int16.tobytes()
                    except ImportError:
                        self._log("需要numpy库来转换音频格式，请安装: pip install numpy")
                        output_audio_bytes = None
                    # 播放Float32：直接把原始数据放入队列
                    if self.is_playing:
                        self.audio_queue.put(audio_data)
                        # print(f"🔊 入队播放(Float32): {len(audio_data)} bytes")

                # 保存输出音频
                if self.output_audio_writer and output_audio_bytes:
                    try:
                        self.output_audio_writer.writeframes(output_audio_bytes)
                    except Exception as e:
                        self._log(f"保存输出音频失败: {e}")
            except Exception as e:
                self._log(f"播放入队失败: {e}")
        
        # 处理完整响应
        elif message_type == 'SERVER_FULL_RESPONSE':
            event = response.get('event')
            payload_msg = response.get('payload_msg', {})
            
            # Event 450 - ASRInfo: 识别出首字，用于打断客户端播报
            if event == 450:
                # print("👤 检测到用户开始说话")
                # print("检测到用户说话当前时间", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
                self.is_user_querying = True
                self.last_text_time = time.time()
                # 用户开始说话时，清空之前的回复文本（新的对话轮次）
                self.current_reply_text = ""
                # 立即打断本地TTS播放：清空播放队列并尝试停流
                try:
                    while not self.audio_queue.empty():
                        try:
                            self.audio_queue.get_nowait()
                        except queue.Empty:
                            break
                    if self.output_stream and self.output_stream.is_active():
                        self.output_stream.stop_stream()
                        # print("⏹ 已本地停止TTS播放并清空队列（打断）")
                        # print("已本地停止TTS当前时间", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
                except Exception as e:
                    self._log(f"停止输出流失败: {e}")
            
            # Event 451 - ASRResponse: 识别结果
            if event == 451:
                # print("451服务器响应当前时间", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
                if isinstance(payload_msg, dict):
                    results = payload_msg.get('results', [])
                    if results and isinstance(results, list) and len(results) > 0:
                        recognized_text = results[0].get('text', '')
                        is_interim = results[0].get('is_interim', False)

                        if recognized_text and isinstance(recognized_text, str) and recognized_text.strip():
                            # 更新last_text_time，避免超时
                            self.last_text_time = time.time()
                            
                            if not is_interim:
                                # 最终识别结果
                                if recognized_text != self.current_text:
                                    self.current_text = recognized_text
                                    self._log(f"✅ 客户回话: {recognized_text}")
                                    # print("kehu当前时间", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))

                                    # 清空当前回复文本，准备收集新的回复
                                    self.current_reply_text = ""
                                    # 触发业务逻辑处理
                                    if self.on_text_received:
                                        self.on_text_received(recognized_text)
                                    asyncio.create_task(self._async_process_text(recognized_text))
                            else:
                                # 中间结果（实时更新）
                                if recognized_text != self.current_text:
                                    self.current_text = recognized_text
                                    # print(f"🔄 用户识别结果（实时）: {recognized_text}")
            
            # Event 459 - ASREnded: 用户说话结束
            if event == 459:
                # print("👤 用户说话结束")
                # print("用户说话结束当前时间", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
                self.is_user_querying = False
                self.last_text_time = time.time()
                # 恢复本地TTS播放能力（流保持打开状态以便后续写入）
                try:
                    if self.output_stream and not self.output_stream.is_active():
                        self.output_stream.start_stream()
                        # print("▶️ 已恢复TTS播放（用户结束说话）")
                except Exception as e:
                    self._log(f"恢复输出流失败: {e}")
                # self.client.chat_tts_text(is_user_querying=self.is_user_querying,)
            # Event 550 - ChatResponse: 豆包回复文本（可能分多次发送）
            if event == 550:
                if isinstance(payload_msg, dict):
                    reply_text = payload_msg.get('content', '')
                    if reply_text and isinstance(reply_text, str) and reply_text.strip():
                        # 累加回复文本（Event 550可能分多次发送）
                        self.current_reply_text += reply_text
                        # print(f"✅ 豆包回复文本（部分）: {reply_text}")

            # Event 559 - ChatEnded: 豆包回复文本结束
            if event == 559:

                # 保存完整的回复文本
                if self.current_reply_text.strip():
                    complete_reply = self.current_reply_text.strip()
                    self._log(f"✅ AI完整回复: {complete_reply}")
                    # print("AI回复当前时间", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
                    if "短信" in complete_reply or "发" in complete_reply or "加微信" in complete_reply:
                        self.sms_doubao_flag = True
                        self._log("短信回复标志已经设置成功True")
                    store_chat_data(self.session_id_for_round, sender=self.device_phone, receiver=self.called_phone, content=complete_reply)
                    if "再见" in complete_reply or "拜拜" in complete_reply or "生活愉快" in complete_reply:
                        asyncio.create_task(self._delay_set_guaduan_phone())

                    # 保存回复记录
                    reply_record = {
                        "session_id": self.session_id_for_round,
                        "user_text": self.current_text,  # 对应的用户识别文本
                        "doubao_text": complete_reply,
                        "timestamp": time.time()
                    }

                    self.all_replies.append(reply_record)

                    # 这里可以调用保存聊天记录的接口
                    # await self._save_chat_record(reply_record)

                    # 清空当前回复文本，准备收集下一次回复
                    self.current_reply_text = ""
            
            # Event 350 - TTSSentenceStart: TTS开始（可能包含text字段）
            if event == 350:
                if isinstance(payload_msg, dict):
                    tts_text = payload_msg.get('text', '')
                    tts_type = payload_msg.get('tts_type', 'default')
                    if tts_text and isinstance(tts_text, str) and tts_text.strip():
                        self._log(f"🎤 TTS开始 ({tts_type}): {tts_text}")
            
            # Event 351 - TTSSentenceEnd: TTS分句结束（payload为空）
            if event == 351:
                pass  # 无需处理
            
            # Event 352 - TTSResponse: 返回音频数据（二进制，已通过SERVER_ACK处理）
            if event == 352:
                pass  # 音频数据已在SERVER_ACK中处理
            
            # Event 359 - TTSEnded: TTS播放结束
            if event == 359:
                self._log("🎵 TTS播放结束")
                if self.hello_playing:
                    self.hello_streaming_done = True
                    # 立即检查是否可以恢复
                    if not self.is_playing or self.audio_queue.empty():
                        self._finish_hello_playback()
                    else:
                        # 如果播放器在运行，启动一个兜底的异步恢复任务（3秒后强制恢复）
                        asyncio.create_task(self._delayed_finish_hello())


            # Event 152/153 - SessionFinished/SessionFailed: 会话结束
            if event == 152 or event == 153:
                self._log(f"会话结束 (Event {event})")
                self.running = False


    async def _delayed_finish_hello(self):
        """延迟恢复开场白播放状态（兜底机制）

        等待2秒后，如果仍在开场白状态，强制恢复正常模式
        这是一个兜底机制，确保即使播放线程检测失败也能恢复
        """
        # 确保设备上下文正确
        set_device_context(self.device_phone)
        await asyncio.sleep(2.0)
        # 3秒后，如果仍在开场白播放或流式传输状态，强制恢复
        if self.hello_playing:
            # print("⏰ 开场白播放超时，强制恢复正常模式（兜底机制）")
            self._finish_hello_playback()

    async def _async_process_text(self, text: str):
        """异步处理识别到的文本（业务逻辑）"""
        # 确保设备上下文正确
        set_device_context(self.device_phone)
        try:
            await asyncio.to_thread(self._process_text, text)
        except Exception as e:
            self._log(f"异步处理文本失败: {e}")
    
    def _process_text(self, text: str):
        """处理识别到的文本（业务逻辑）"""
        # 确保设备上下文正确
        set_device_context(self.device_phone)
        try:
            # print(f"处理文本: {text}")
            
            # 命中拦截关键词
            if any(kw in text for kw in BLOCK_PHRASES):
                self._init_redis()
                if self.redis_client is not None:
                    try:
                        self.redis_client.set(self.redis_key, "novice")
                    except Exception:
                        pass
                self._log("命中拦截关键词，已置为 novice 并跳过业务调用")
                self.running = False
                return

            # 处理短信命中关键字进行发送
            if self.sms_doubao_flag:
                # if "可以" in text or "好的" in  text or "嗯" in text or "行" in  text or "方便" in text or "发" in text:
                #     pass
                    # print("命中客户短信成功")
                if is_sending_sms(text)["result"]=="需要":
                    send_sms(self.called_phone, self.device_phone)
                    self._log("✅ 已发送短信给用户")
                self.sms_doubao_flag = False
            """
            调用保存聊天记录的接口
            将聊天记录保存到服务端
            {
                session_id:"",
                sender:"",
                customer_id,
                doubao_text
            }
            """
            store_chat_data(self.session_id_for_round, self.called_phone, self.device_phone, text)

        except Exception as e:
            self._log(f"处理文本时出错: {e}")

    
    async def _receive_loop(self):
        """接收服务器响应的循环"""
        # 确保设备上下文正确
        set_device_context(self.device_phone)
        try:
            while self.running:
                try:
                    if self.client is None or self.client.ws is None:
                        self._log("客户端或WebSocket为空，退出接收循环")
                        break
                    try:
                        response = await asyncio.wait_for(
                            self.client.receive_server_response(),
                            timeout=1.0
                        )
                        await self._handle_server_response(response)
                    except asyncio.TimeoutError:
                        # 超时是正常的，继续循环
                        continue
                    except Exception as e:
                        self._log(f"接收消息错误: {e}")
                        import traceback
                        traceback.print_exc()
                        # 如果是连接关闭错误，退出循环
                        if "recv" in str(e).lower() or "closed" in str(e).lower():
                            self._log("WebSocket连接已关闭，退出接收循环")
                            break
                except Exception as e:
                    self._log(f"接收循环内部错误: {e}")
                    if not self.running:
                        break
        except Exception as e:
            self._log(f"接收循环错误: {e}")
            import traceback
            traceback.print_exc()


    # 添加新的异步方法
    async def _delay_set_guaduan_phone(self):
        """延迟8秒后设置挂断状态"""
        # 确保设备上下文正确
        set_device_context(self.device_phone)
        await asyncio.sleep(8)
        set_guaduan_phone(self.device_phone, "guaduan")
        self._log("已成功设置挂断状态")

    async def _send_audio_loop(self, input_stream):
        """发送音频数据的循环"""
        # 确保设备上下文正确
        set_device_context(self.device_phone)
        audio_chunk_count = 0
        
        try:
            while self.running:
                try:
                    # 检查客户端和WebSocket是否有效
                    if self.client is None or self.client.ws is None:
                        self._log("客户端或WebSocket为空，退出音频发送循环")
                        break

                    # 如果开场白仍在播放或尚未完全结束，读取但丢弃音频数据（清空缓冲区，避免积累）
                    if self.hello_playing or not self.hello_streaming_done:
                        self.last_text_time = time.time()
                        try:
                            # 读取音频数据以清空缓冲区，但不发送也不保存
                            if self.input_stream is not None:
                                _ = self.input_stream.read(CHUNK, exception_on_overflow=False)
                        except Exception:
                            pass
                        await asyncio.sleep(0.01)
                        continue
                    
                    # 检查Redis开关
                    self._init_redis()
                    if self.redis_client is not None:
                        try:
                            field_val = self.redis_client.get(self.redis_key)
                            current_status = None
                            if isinstance(field_val, str) and "_" in field_val:
                                parts = field_val.rsplit("_", 1)
                                if len(parts) == 2:
                                    current_status = parts[1]
                            if current_status != VOICE_CONNECTED_VALUE:
                                self._log(f"设备 {self.device_phone} 检测到开关关闭（状态={current_status}），停止识别并结束会话...")
                                self.running = False
                                break
                        except Exception:
                            pass
                    
                    # 检查超时（仅在非用户说话期间检查）
                    if not self.is_user_querying:
                        if time.time() - self.last_text_time > INACTIVITY_TIMEOUT_SECONDS:
                            self._log(f"设备 {self.device_phone} 超过 {INACTIVITY_TIMEOUT_SECONDS} 秒未识别到文本或用户活动，自动关闭")
                            self._init_redis()
                            if self.redis_client is not None:
                                try:
                                    self.redis_client.set(self.redis_key, "novice", ex=60*60*12)
                                    set_guaduan_phone(self.device_phone, "guaduan")
                                except Exception:
                                    pass
                            self.running = False
                            break
                    
                    # 读取音频数据
                    try:
                        # 检查输入流是否有效
                        if input_stream is None:
                            self._log("错误: 输入流为None，退出音频发送循环")
                            self.running = False
                            break
                        
                        audio_data = input_stream.read(CHUNK, exception_on_overflow=False)
                        if not audio_data or len(audio_data) == 0:
                            self._log("警告: 读取到空音频数据")
                            await asyncio.sleep(0.01)
                            continue
                        
                        # 检查客户端是否有效
                        if self.client is None:
                            self._log("错误: 客户端为None，无法发送音频")
                            self.running = False
                            break
                        
                        if self.client.ws is None:
                            self._log("错误: WebSocket为None，无法发送音频")
                            self.running = False
                            break
                        
                        # 发送音频数据到豆包
                        await self.client.task_request(bytes(audio_data))
                        audio_chunk_count += 1
                        
                        # 保存输入音频（客户说的音频）
                        if self.input_audio_writer:
                            try:
                                self.input_audio_writer.writeframes(audio_data)
                            except Exception as e:
                                self._log(f"保存输入音频失败: {e}")

                        # 计算合适的sleep时间：chunk_size / sample_rate
                        # 3200帧 / 16000Hz = 0.2秒，但实际发送更快一些，sleep 0.01即可
                        await asyncio.sleep(0.01)
                    except Exception as e:
                        self._log(f"读取或发送音频失败: {e}")
                        import traceback
                        traceback.print_exc()
                        await asyncio.sleep(0.1)
                
                except Exception as e:
                    self._log(f"发送音频循环错误: {e}")
                    import traceback
                    traceback.print_exc()
                    await asyncio.sleep(0.1)
        except Exception as e:
            self._log(f"发送音频循环异常: {e}")
            import traceback
            traceback.print_exc()


    
    def start_recognition(self, customer_info: Optional[dict] = None):
        """启动识别（同步接口）
        
        Args:
            customer_info: 客户信息字典，包含 customer_name 等字段，用于动态配置提示词
        """
        self._setup_event_loop()
        
        # 如果传入了客户信息，使用它；否则使用之前设置的
        if customer_info:
            self.customer_info = customer_info
        
        async def _start():
            # 确保设备上下文正确
            set_device_context(self.device_phone)

            # 创建豆包客户端
            self.session_id = str(uuid.uuid4())
            ws_config = doubao_config.ws_connect_config.copy()
            
            # 从ConfigManager动态获取配置
            import copy
            start_session_req = copy.deepcopy(doubao_config.get_start_session_req())
            start_session_req.pop("model_speaker")
            # 动态构建提示词，包含客户信息
            system_role = start_session_req["dialog"]["system_role"]
            if self.customer_info:
                customer_name = self.customer_info.get("customer_name", "")
                gender = self.customer_info.get("gender", "")
                customer_age = self.customer_info.get("customer_age", "")
                customer_situation = self.customer_info.get("customer_situation", "")
                # 在提示词中加入客户信息
                if customer_name != "客户":
                    system_role = f"""
                    重要信息：
                    - 当前客户姓名：{customer_name}
                    - 当前客户性别：{gender}
                    - 当前客户年龄：{customer_age}
                    - 当前客户情况：{customer_situation}
                    - 当前时间：{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}
                    - 请在与客户交流时使用客户姓名，使对话更个性化。
                    
                    """+system_role
                else:
                    system_role = f"""
                    重要信息：
                    - 当前客户姓名：未知
                    - 当前时间：{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}
                    
                    """+system_role
            
            # 更新配置中的system_role
            try:
                if start_session_req["dialog"]["extra"].get("model","")=="SC":
                    start_session_req["dialog"]["character_manifest"] = system_role
                    start_session_req["dialog"].pop("system_role")
                else:
                    start_session_req["dialog"]["system_role"] = system_role
            except Exception:
                self._log("警告: 配置中的system_role字段可能不存在，请检查配置")
                start_session_req["dialog"]["system_role"] = system_role
            # 显式开启双工打断相关能力（字段名依据官方文档/实际可用为准）
            try:
                extra = start_session_req.setdefault("dialog", {}).setdefault("extra", {})
                # 常见可用的开关名称（不同版本可能有所差异，尽量全部打开）
                extra["enable_duplex"] = True
                extra["barge_in"] = True
                extra["interrupt_tts_on_asr"] = True
                # 可选：降低端点静默阈值，加快端点判断
                if "eos_silence_timeout" not in extra:
                    extra["eos_silence_timeout"] = 500  # ms
                if "end_smooth_silence_proportion" not in extra:
                    extra["end_smooth_silence_proportion"] = 0.5
            except Exception:
                pass
            
            # 从配置中获取recv_timeout
            recv_timeout = start_session_req.get("dialog", {}).get("extra", {}).get("recv_timeout", 10)
            input_mod = start_session_req.get("dialog", {}).get("extra", {}).get("input_mod", "audio")
            
            # 创建客户端
            self.client = RealtimeDialogClient(
                config=ws_config,
                session_id=self.session_id,
                output_audio_format="pcm",
                mod=input_mod,
                recv_timeout=recv_timeout,
                start_session_req=start_session_req  # 传入自定义配置
            )
            # print("使用动态配置：", start_session_req)
            # 连接WebSocket
            await self.client.connect()
            self._log("WebSocket连接成功")
            
            # 创建音频保存目录（使用session_id_for_round，如果没有则使用session_id）
            save_session_id = self.session_id_for_round
            if save_session_id:
                self.audio_save_dir = os.path.join("audio_recordings", save_session_id)
                os.makedirs(self.audio_save_dir, exist_ok=True)
                self._log(f"音频保存目录: {self.audio_save_dir}")
                
                # 初始化音频文件（输入：客户说的音频）
                input_audio_path = os.path.join(self.audio_save_dir, "input_audio.wav")
                self.input_audio_writer = wave.open(input_audio_path, 'wb')
                self.input_audio_writer.setnchannels(CHANNELS)
                self.input_audio_writer.setsampwidth(pyaudio.PyAudio().get_sample_size(FORMAT))
                self.input_audio_writer.setframerate(RATE)
                # print(f"输入音频保存路径: {input_audio_path}")
                
                # 初始化音频文件（输出：豆包TTS音频）
                output_audio_path = os.path.join(self.audio_save_dir, "output_audio.wav")
                self.output_audio_writer = wave.open(output_audio_path, 'wb')
                self.output_audio_writer.setnchannels(1)  # 单声道
                self.output_audio_writer.setsampwidth(2)  # 16位 = 2字节（Float32会转换为Int16）
                self.output_audio_writer.setframerate(24000)  # 豆包输出采样率24000Hz
                # print(f"输出音频保存路径: {output_audio_path}")

            t = time.time()
            current_hour = time.localtime(t).tm_hour

            # 根据小时数判断时间段
            if 6 <= current_hour < 12:
                greeting = "上午好"
            elif 12 <= current_hour < 18:
                greeting = "下午好"
            else:
                greeting = "晚上好"
            kcb_template = start_session_req.get("kcb", "喂{greeting}")

            if self.customer_info.get("gender", "") == "男":
                customer_name = self.customer_info.get("customer_name", "")[:1]+"先生"
            elif self.customer_info.get("gender", "") == "女":
                customer_name = self.customer_info.get("customer_name", "")[:1]+"女士"
            elif self.customer_info.get("customer_name", "") == "客户":
                customer_name =  self.customer_info.get("customer_name")
            else:
                customer_name=self.customer_info.get("customer_name", "")[:1]+"总"
            # 准备所有可能的占位符值
            format_kwargs = {
                "greeting": greeting,
                "customer_name": customer_name
            }

            # 安全地格式化字符串，只替换模板中存在的占位符
            try:
                kcb_content = kcb_template.format(**format_kwargs)
            except KeyError as e:
                kcb_content = kcb_template
                for key, value in format_kwargs.items():
                    kcb_content = kcb_content.replace(f"{{{key}}}", str(value))
            # 在发送开场白时：
            self.hello_playing = True
            self.hello_streaming_done = False
            self._log(f"开场白：{kcb_content}")
            await self.client.say_hello(content=f"{kcb_content}")

            # 启动播放线程（与官方demo一致的播放模型）
            if self.output_stream and not self.is_playing:
                self.is_playing = True
                # 保存设备上下文给播放线程使用
                device_phone = self.device_phone
                def _player():
                    # 在播放线程中设置设备上下文
                    set_device_context(device_phone)
                    idle_loops = 0
                    while self.is_playing:
                        try:
                            data = self.audio_queue.get(timeout=1.0)
                            idle_loops = 0
                            if not data:
                                continue
                            # 若用户正在说话则丢弃，保障打断（但如果正在播放开场白，则继续播放）
                            if self.is_user_querying and not self.hello_playing:
                                continue
                            # 确保输出流处于活动状态
                            if self.output_stream and not self.output_stream.is_active():
                                try:
                                    self.output_stream.start_stream()
                                except Exception:
                                    pass
                            # 写入播放
                            try:
                                self.output_stream.write(data)
                            except Exception:
                                time.sleep(0.01)
                        except queue.Empty:
                            idle_loops += 1
                            # 如果开场白流式传输已结束，且队列已空闲，立即恢复（不需要等待2次）
                            if self.hello_playing and self.hello_streaming_done and idle_loops >= 1:
                                self._finish_hello_playback()
                            time.sleep(0.05)
                        except Exception:
                            time.sleep(0.05)
                self.player_thread = threading.Thread(target=_player, daemon=True)
                self.player_thread.start()
            
        self._run_in_loop(_start())
    
    def start_microphone_for_device(self):
        """为指定设备启动麦克风录音"""
        self._init_audio()
        
        input_device_index = self.device_index
        if input_device_index is None:
            try:
                device_count = self.audio.get_device_count()
                for i in range(device_count):
                    info = self.audio.get_device_info_by_index(i)
                    if int(info.get('maxInputChannels', 0)) > 0:
                        input_device_index = i
                        break
            except Exception:
                input_device_index = None
        
        try:
            self.input_stream = self.audio.open(
                format=FORMAT,
                channels=CHANNELS,
                rate=RATE,
                input=True,
                output=False,
                input_device_index=input_device_index,
                frames_per_buffer=CHUNK
            )
            
            # 打开输出流（豆包输出Float32格式，24000采样率）
            if self.output_device_index is not None:
                try:
                    self.output_stream = self.audio.open(
                        format=pyaudio.paFloat32,
                        channels=1,
                        rate=24000,  # 豆包输出采样率
                        output=True,
                        output_device_index=self.output_device_index,
                        frames_per_buffer=3200
                    )
                    # print(f"输出流已打开，设备: {self.output_device_index}")
                except Exception as e:
                    self._log(f"打开输出流失败: {e}，尝试使用Int16格式")
                    # 如果Float32不支持，尝试Int16
                    try:
                        self.output_stream = self.audio.open(
                            format=pyaudio.paInt16,
                            channels=1,
                            rate=24000,
                            output=True,
                            output_device_index=self.output_device_index,
                            frames_per_buffer=3200
                        )
                        self.output_format = 'int16'  # 标记需要转换
                    except Exception as e2:
                        self._log(f"使用Int16格式也失败: {e2}")
                        self.output_stream = None
            
            device_name = "默认设备"
            if input_device_index is not None:
                try:
                    info = self.audio.get_device_info_by_index(input_device_index)
                    device_name = info.get('name', f'设备_{input_device_index}')
                except:
                    device_name = f'设备_{input_device_index}'
        except Exception as e:
            self._log(f"设备 {self.device_phone} 麦克风启动失败: {e}")
            raise e
    
    def send_audio(self):
        """发送音频数据到识别器（同步接口）"""
        self._log("开始录音和识别...请开始说话，说完一句话后会有结果输出")
        
        # 确保input_stream已打开
        if self.input_stream is None:
            self._log("错误: 输入流未打开，无法发送音频")
            return
        
        self.running = True
        self.last_text_time = time.time()
        
        async def _run():
            # 确保设备上下文正确
            set_device_context(self.device_phone)

            # 启动接收循环
            receive_task = asyncio.create_task(self._receive_loop())
            # 启动发送循环
            send_task = asyncio.create_task(self._send_audio_loop(self.input_stream))
            # print("接收和发送任务已启动")

            # 与官方demo一致：保持主循环存活，由内部状态控制退出
            try:
                while self.running:
                    await asyncio.sleep(0.1)
            except asyncio.CancelledError:
                pass
            except Exception as e:
                self._log(f"运行任务时出错: {e}")
                import traceback
                traceback.print_exc()
            finally:
                # 结束时取消子任务
                for t in (receive_task, send_task):
                    if not t.done():
                        t.cancel()
                        try:
                            await t
                        except Exception:
                            pass
                self._log("音频发送任务已结束")
        
        try:
            self._run_in_loop(_run())
        except Exception as e:
            self._log(f"运行音频任务失败: {e}")
            import traceback
            traceback.print_exc()
    
    def stop(self):
        """停止识别和录音"""
        # print("开始停止识别器...")
        self.running = False
        # 停止播放线程
        if self.is_playing:
            self.is_playing = False
            try:
                if self.player_thread and self.player_thread.is_alive():
                    self.player_thread.join(timeout=1.5)
            except Exception:
                pass
        
        # 关闭音频保存文件
        if self.input_audio_writer:
            try:
                self.input_audio_writer.close()
                self.input_audio_writer = None
                # print("输入音频文件已保存")
            except Exception as e:
                self._log(f"关闭输入音频文件失败: {e}")
        
        if self.output_audio_writer:
            try:
                self.output_audio_writer.close()
                self.output_audio_writer = None
                # print("输出音频文件已保存")
            except Exception as e:
                self._log(f"关闭输出音频文件失败: {e}")
        
        async def _stop():
            # 确保设备上下文正确
            set_device_context(self.device_phone)

            if self.client:
                try:
                    # 先尝试正常结束会话
                    try:
                        await asyncio.wait_for(self.client.finish_session(), timeout=2.0)
                    except asyncio.TimeoutError:
                        self._log("结束会话超时")
                    except Exception as e:
                        self._log(f"结束会话失败: {e}")
                    
                    await asyncio.sleep(0.1)
                    
                    # 尝试结束连接
                    try:
                        await asyncio.wait_for(self.client.finish_connection(), timeout=2.0)
                    except asyncio.TimeoutError:
                        self._log("结束连接超时")
                    except Exception as e:
                        self._log(f"结束连接失败: {e}")
                    
                    await asyncio.sleep(0.1)
                    
                    # 关闭WebSocket
                    try:
                        await asyncio.wait_for(self.client.close(), timeout=2.0)
                    except asyncio.TimeoutError:
                        self._log("关闭WebSocket超时")
                    except Exception as e:
                        self._log(f"关闭WebSocket失败: {e}")
                except Exception as e:
                    self._log(f"关闭客户端失败: {e}")
                finally:
                    self.client = None
        
        if self.loop and self.loop.is_running():
            try:
                # 使用较短的超时，避免等待太久
                self._run_in_loop(_stop(), timeout=5)
            except TimeoutError:
                self._log("停止超时，强制关闭")
            except Exception as e:
                self._log(f"停止时出错: {e}")
        
        if self.input_stream:
            try:
                self.input_stream.stop_stream()
                self.input_stream.close()
            except Exception:
                pass
            self.input_stream = None
        
        if self.output_stream:
            try:
                self.output_stream.stop_stream()
                self.output_stream.close()
            except Exception:
                pass
            self.output_stream = None
        
        self._log("已停止录音和识别")
    
    def disconnect_ws_only(self):
        """仅断开WebSocket连接，不停止其他流程（用于临时停止AI回复）
        
        此方法只关闭WS连接，保持recognizer实例和其他状态不变，
        设备可以继续监听，等待恢复。
        """
        self._log("正在断开WebSocket连接（仅断开AI回复）...")
        # 设置标志，停止接收和发送循环，但不影响recognizer实例
        self.running = False
        
        # 停止播放线程
        if self.is_playing:
            self.is_playing = False
            try:
                if self.player_thread and self.player_thread.is_alive():
                    self.player_thread.join(timeout=1.5)
            except Exception:
                pass
        
        # 清空音频队列
        try:
            while not self.audio_queue.empty():
                try:
                    self.audio_queue.get_nowait()
                except queue.Empty:
                    break
        except Exception:
            pass
        
        # 只关闭WebSocket连接，不关闭音频流和其他资源
        async def _disconnect_ws():
            # 确保设备上下文正确
            set_device_context(self.device_phone)
            
            if self.client:
                try:
                    # 先尝试正常结束会话
                    try:
                        await asyncio.wait_for(self.client.finish_session(), timeout=2.0)
                    except asyncio.TimeoutError:
                        self._log("结束会话超时")
                    except Exception as e:
                        self._log(f"结束会话失败: {e}")
                    
                    await asyncio.sleep(0.1)
                    
                    # 尝试结束连接
                    try:
                        await asyncio.wait_for(self.client.finish_connection(), timeout=2.0)
                    except asyncio.TimeoutError:
                        self._log("结束连接超时")
                    except Exception as e:
                        self._log(f"结束连接失败: {e}")
                    
                    await asyncio.sleep(0.1)
                    
                    # 关闭WebSocket
                    try:
                        await asyncio.wait_for(self.client.close(), timeout=2.0)
                    except asyncio.TimeoutError:
                        self._log("关闭WebSocket超时")
                    except Exception as e:
                        self._log(f"关闭WebSocket失败: {e}")
                except Exception as e:
                    self._log(f"关闭客户端失败: {e}")
                finally:
                    self.client = None
        
        if self.loop and self.loop.is_running():
            try:
                self._run_in_loop(_disconnect_ws(), timeout=5)
            except TimeoutError:
                self._log("断开WS超时，强制关闭")
            except Exception as e:
                self._log(f"断开WS时出错: {e}")
        
        self._log("WebSocket连接已断开（设备继续监听）")
    
    def get_all_replies(self):
        """获取所有完整的豆包回复记录
        
        Returns:
            list: 所有回复记录列表，每个记录包含：
                - session_id: 会话ID
                - user_text: 用户识别文本
                - doubao_text: 豆包回复文本
                - timestamp: 时间戳
        """
        return self.all_replies.copy()
    
    def get_last_reply(self):
        """获取最后一次完整的豆包回复文本
        
        Returns:
            str: 最后一次完整回复文本，如果没有则返回空字符串
        """
        if self.all_replies:
            return self.all_replies[-1].get('doubao_text', '')
        return ''
    
    def clear_replies(self):
        """清空所有回复记录"""
        self.all_replies = []
        self.current_reply_text = ""
    
    def close(self):
        """程序最终退出时释放底层音频资源"""
        self.stop()
        try:
            if self.audio:
                self.audio.terminate()
                self.audio = None
        except Exception:
            pass
        
        # 停止事件循环
        if self.loop and self.loop.is_running():
            try:
                self.loop.call_soon_threadsafe(self.loop.stop)
            except:
                pass
            if self.loop_thread:
                self.loop_thread.join(timeout=2)
                self.loop = None
                self.loop_thread = None
                self.say_hello_over_event = None
    

    
    def wait_until_dialing_for_device(self, stop_flag=None):
        """设备专用等待拨号方法
        
        Args:
            stop_flag: 可选的停止标志函数，如果返回True则停止等待
        """
        self._init_redis()
        if self.redis_client is None:
            return
        while True:
            # 检查停止标志
            if stop_flag and stop_flag():
                self._log(f"设备 {self.device_phone} 等待拨号被中断")
                return
            try:
                field_val = self.redis_client.get(self.redis_key)
                current_status = None
                if isinstance(field_val, str) and "_" in field_val:
                    parts = field_val.rsplit("_", 1)
                    if len(parts) == 2:
                        current_status = parts[1]
                if current_status == "dialing":
                    return
            except Exception:
                pass
            time.sleep(0.5)


